package com.shujia.useraction

import com.shujia.useraction.UserActionIPTV.log
import com.shujia.utils.{Constants, SparkMain, SparkTool}
import org.apache.spark.internal.Logging
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

class UserActionMobile(day: String) extends UserActionBase {

  import spark.implicits._
  import org.apache.spark.sql.functions._


  override def load(path: String, day: String): DataFrame = {
    spark
      .read
      .format("csv")
      .option("sep", "|")
      .schema("videoname String, userid String, sn String, start_time String, end_time String, albumid String, albumname String, videoid String, c1 String, csource String, trace_id String, fee_type String, istry String, iscon String, start_posi String")
      .load(s"$path/$day")
  }

  override def process(): Unit = {
    //    df = load(Constants.USER_ACTION_BASE_MOBILE_PATH, day)

    val mobileRDD: RDD[String] = spark.sparkContext.textFile(s"${Constants.USER_ACTION_BASE_MOBILE_PATH}/$day")
    df = mobileRDD
      .map(_.split("\\|"))
      .filter(_.length == 15)
      .map(arr => (arr(0), arr(1), arr(2), arr(3), arr(4), arr(5), arr(6), arr(7), arr(8), arr(9), arr(10), arr(11), arr(12), arr(13), arr(14)))
      .toDF("videoname", "userid", "sn", "start_time", "end_time", "albumid", "albumname", "videoid", "c1", "csource", "trace_id", "fee_type", "istry", "iscon", "start_posi")
      .select(upper(md5($"userid")) as "uid", $"albumid" as "itemId", $"albumname" as "itemName", $"start_time", $"end_time")
      .withColumn("play_time", unix_timestamp($"end_time") - unix_timestamp($"start_time"))

    df.show()

  }

  override def save(): Unit = {

    df.write
      .format(Constants.USER_ACTION_MOBILE_FORMAT)
      .mode(SaveMode.Overwrite)
      .save(s"${Constants.USER_ACTION_MOBILE_PATH}/$day")
  }

  override def load(): DataFrame = {
    spark
      .read
      .format(Constants.USER_ACTION_MOBILE_FORMAT)
      .load(s"${Constants.USER_ACTION_MOBILE_PATH}/$day")
  }
}

object UserActionMobile extends SparkMain {
  override def run(day: String): Unit = {
    val userActionMobile: UserActionMobile = UserActionMobile(day)

    userActionMobile.process()
    userActionMobile.save()


  }

  def apply(day: String): UserActionMobile = new UserActionMobile(day)
}